package com.aiwebops.spark;

import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;

public class SecondarySortApp {
    public static void main(String[] args) throws Exception {
        SparkConf conf = new SparkConf()
                .setAppName("SecondarySortApp")
                .setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        //去掉不必要的输出信息
        sc.setLogLevel("WARN");

        // 调用 textFile() 方法，读取日志文件，这里指定本地磁盘文件，您也可以指定是 HDFS 上的文件

        JavaRDD<String> tf = sc.textFile(
                "src/main/resources/access_20170504.log");

        // 调用 mapTfRDD2Pair 方法 将 tf 映射为键值对
        JavaPairRDD<String, cmbInfo> tfPairRDD =
                mapTfRDD2Pair(tf);

        // 获取每个手机号的总上行流量、总下行流量、最早报告时间戳
        JavaPairRDD<String, cmbInfo> agTfPairRDD =
                aggregateByDeviceID(tfPairRDD);
        //聚合，封装的 RDD 作为 key,手机号作为值
        JavaPairRDD<SecondarySortKey, String> tfSortRDD =
                mapRDDKey2SortKey(agTfPairRDD);

        // 依次按照上行流量、下行流量以及报告时间戳倒序排序
        JavaPairRDD<SecondarySortKey, String> sortedTfRDD =
                tfSortRDD.sortByKey(false);
        // 根据您的需要获得输出，这里仅显示前 8 行
        List<Tuple2<SecondarySortKey, String>> dtTop =
                sortedTfRDD.take(8);
        System.out.println("============ result ============");
        for (Tuple2<SecondarySortKey, String> dt : dtTop) {
            System.out.println("phoneNum " + dt._2 + ", " + dt._1);
        }

        // 记住执行结束关闭资源
        sc.close();
    }

    //mapTfRDD2Pair 方法，封装键值对
    private static JavaPairRDD<String, cmbInfo> mapTfRDD2Pair(
            JavaRDD<String> accessLogRDD) {
        return accessLogRDD.mapToPair(new PairFunction<String, String, cmbInfo>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<String, cmbInfo> call(String lines)
                    throws Exception {
                // 根据数据格式进行切分
                String[] split = lines.split("\t");

                // 获取切分的字段
                long reportTime = Long.valueOf(split[0]);
                String phoneNum = split[1];
                long upPayLoad = Long.valueOf(split[2]);
                long downPayload = Long.valueOf(split[3]);

                // 创建 cmbInfo 对象，有参构造 ，将上行流量、下行流量，报告时间戳封装为自定义的可序列化对象
                cmbInfo dataInfo = new cmbInfo(reportTime,
                        upPayLoad, downPayload);

                return new Tuple2<String, cmbInfo>(phoneNum, dataInfo);
            }

        });
    }

    //根据手机号进行聚合，依次按照上行流量、下行流量以及报告时间戳倒序排序
    private static JavaPairRDD<String, cmbInfo> aggregateByDeviceID(
            JavaPairRDD<String, cmbInfo> accessLogPairRDD) {
        return accessLogPairRDD.reduceByKey(new Function2<cmbInfo, cmbInfo, cmbInfo>() {

            private static final long serialVersionUID = 1L;

            @Override
            public cmbInfo call(cmbInfo d1, cmbInfo d2)
                    throws Exception {
                long reportTime = d1.getReportTime() < d2.getReportTime() ?
                        d1.getReportTime() : d2.getReportTime();
                long upPayLoad = d1.getUpPayLoad() + d2.getUpPayLoad();
                long downPayload = d1.getDownPayload() + d2.getDownPayload();

                cmbInfo accessLogInfo = new cmbInfo();
                accessLogInfo.setReportTime(reportTime);
                accessLogInfo.setUpPayLoad(upPayLoad);
                accessLogInfo.setDownPayload(downPayload);

                return accessLogInfo;
            }

        });
    }

    //二次排序，手机号作为值
    private static JavaPairRDD<SecondarySortKey, String> mapRDDKey2SortKey(
            JavaPairRDD<String, cmbInfo> aggrAccessLogPairRDD) {
        return aggrAccessLogPairRDD.mapToPair(

                new PairFunction<Tuple2<String, cmbInfo>, SecondarySortKey, String>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<SecondarySortKey, String> call(
                            Tuple2<String, cmbInfo> tuple) throws Exception {
                        // 获取元祖里的数据
                        String phoneNum = tuple._1;
                        cmbInfo accessLogInfo = tuple._2;

                        // 封装为二次排序 key
                        SecondarySortKey accessLogSortKey = new SecondarySortKey(
                                accessLogInfo.getUpPayLoad(),
                                accessLogInfo.getDownPayload(),
                                accessLogInfo.getReportTime());

                        // 返回新的 Tuple
                        return new Tuple2<SecondarySortKey, String>(accessLogSortKey, phoneNum);
                    }

                });
    }

}

